package org.fastsyncer.core.service.impl;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import org.apache.commons.lang.StringUtils;
import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.constant.EventConstant;
import org.fastsyncer.common.constant.MappingConstant;
import org.fastsyncer.common.constant.RestConstant;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.common.entity.Page;
import org.fastsyncer.core.service.ManagerService;
import org.fastsyncer.core.util.MappingTaskUtil;
import org.fastsyncer.manager.framework.DataFactory;
import org.fastsyncer.manager.framework.ManagerFactory;
import org.fastsyncer.manager.log.LogFactory;
import org.fastsyncer.manager.monitor.MonitorFactory;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * 驱动业务操作API,包含驱动新增、修改、删除、查询、启动、停止
 * @author AE86
 */
@Service("ManagerService")
public final class ManagerServiceImpl implements ManagerService {

    private static final Logger logger = LoggerFactory.getLogger(ManagerServiceImpl.class);

    @Autowired
    private DriverEventServiceImpl event;

    // 同步驱动日志操作事务，防止重复提交日志
    private volatile Map<String, Boolean> transaction = new ConcurrentHashMap<>();

    private DataFactory data = DataFactory.getInstance();

    private LogFactory log = LogFactory.getInstance();

    private ManagerFactory manager = ManagerFactory.getInstance();

    private MonitorFactory monitor = MonitorFactory.getInstance();

    @Override
    public String add(String mapping) throws Exception {
        MappingTask mappingTask = MappingTaskUtil.parse(mapping);
        String id = mappingTask.getId();
        boolean store = data.saveMapping(id, mappingTask);
        if (!store) {
            logger.error("Add driver failed!");
            throw new Exception("新增驱动失败!");
        }
        // 触发新增事件
        event.notice(EventConstant.ADD, id);
        return mappingTask.getId();
    }

    @Override
    public String update(String mapping) throws Exception {
        MappingTask mappingTask = MappingTaskUtil.parse(mapping);
        // 当前为启用
        if (!mappingTask.isEnable()) {
            throw new Exception("驱动已被禁用.");
        }

        // 检查驱动是否在运行,如果运行给出提示
        String model = mappingTask.getModel();
        if (MappingConstant.SYNC_ALL.equals(model)) {
            //检查驱动是否处于运行状态
            if (this.isRun(mappingTask.getId())) {
                throw new Exception("驱动正在运行,请先停止.");
            }

            // 获取策略
            Map<String, Map<String, String>> policy = mappingTask.getPolicy();
            Map<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
            // 关闭同步模式下驱动功能
            if (null != po) {
                po.put("enable", "false");
            }
        } else {
            // 获取策略
            Map<String, Map<String, String>> policy = mappingTask.getPolicy();
            Map<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
            // 是否开启功能
            if (null != po) {
                if (Boolean.valueOf(po.get("enable"))) {
                    throw new Exception("驱动正在运行,请先停止.");
                }
            }
        }
        // 更新操作时间
        mappingTask.setUpdateTime(System.currentTimeMillis());

        // 保存驱动配置
        String id = mappingTask.getId();
        boolean save = data.saveMapping(id, mappingTask);
        log.deleteLog(id);
        if (!save) {
            logger.error("Update driver failed!");
            throw new Exception("修改驱动失败!");
        }
        // 触发修改事件
        event.notice(EventConstant.UPDATE, id);
        return "修改驱动成功!";
    }

    @Override
    public String enable(String id) throws Exception {
        MappingTask mappingTask = data.getMapping(id);
        if (mappingTask == null) {
            throw new Exception("驱动不存在.");
        }
        // 修改状态为false,去掉启用时间
        mappingTask.setEnable(true);
        mappingTask.setStartDate(new Date());

        boolean store = data.saveMapping(id, mappingTask);
        if (!store) {
            logger.error("Update driver failed!");
            throw new Exception("修改驱动状态失败!");
        }
        // 触发恢复事件
        event.notice(EventConstant.ENABLE, id);
        return "恢复驱动成功!";
    }

    @Override
    public String disable(String id) throws Exception {
        MappingTask mappingTask = data.getMapping(id);
        if (mappingTask == null) {
            throw new Exception("驱动不存在.");
        }
        //检查驱动是否处于运行状态
        if (this.isRun(id)) {
            throw new Exception("请先停止驱动或等待任务结束.");
        }

        // 获取策略
        Map<String, Map<String, String>> policy = mappingTask.getPolicy();
        Map<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
        // 是否开启功能
        if (po != null) {
            if (Boolean.valueOf(po.get("enable"))) {
                throw new Exception("驱动正在运行,请先停止.");
            }
        }

        // 修改状态为true,设置最新启用时间
        mappingTask.setEnable(false);
        mappingTask.setStartDate(null);

        boolean store = data.saveMapping(id, mappingTask);
        if (!store) {
            logger.error("Update driver failed!");
            throw new Exception("修改驱动状态失败!");
        }
        // 触发禁用事件
        event.notice(EventConstant.DISABLE, id);
        return "禁用驱动成功!";
    }

    @Override
    public String start(String id) throws Exception {
        MappingTask mappingTask = data.getMapping(id);

        // 1、检查驱动
        if (mappingTask == null) {
            throw new IllegalArgumentException("驱动不存在！");
        }

        // 2、检查是否启用
        if (!mappingTask.isEnable()) {
            throw new IllegalArgumentException("驱动状态必须为启用！");
        }

        // 3、如果上一次的任务已经执行完成，则还原同步默认值
        if (mappingTask.getProcess() == mappingTask.getProcessCompleted()) {
            mappingTask.setSuccess(0);
            mappingTask.setFail(0);
            mappingTask.setProcessCompleted(0);
            mappingTask.setRunTime(0L);
        }

        // 分别处理数据迁移和数据同步业务
        String model = mappingTask.getModel();
        switch (model) {
        case MappingConstant.SYNC_ALL:
            // 4、检查同步属性、主键
            this.validateSyncFields(mappingTask);

            // 5、校验同步任务是否正在处理中
            manager.validateTask(id);

            // 6、处理任务
            manager.handle(mappingTask);
            break;
        case MappingConstant.SYNC_INCREMENT:
            // 检查同步属性
            this.validateSyncFields(mappingTask);

            Map<String, Map<String, String>> policy = mappingTask.getPolicy();
            Map<String, String> inc = policy.get(MappingConstant.POLICY_INCREMENT);
            if (inc == null) {
                policy.put(MappingConstant.POLICY_INCREMENT, new HashMap<String, String>());
                inc = policy.get(MappingConstant.POLICY_INCREMENT);
            }
            inc.put("enable", "true");
            boolean store = data.saveMapping(id, mappingTask);
            if (!store) {
                logger.error("Start driver failed!");
                throw new Exception("启动驱动失败!");
            }
            break;
        default:
            break;
        }
        // 触发启动事件
        event.notice(EventConstant.START, id);
        return "驱动启动成功！";
    }

    @Override
    public String stop(String id) throws Exception {
        MappingTask mappingTask = data.getMapping(id);

        // 1、检查驱动
        if (mappingTask == null) {
            throw new IllegalArgumentException("驱动不存在！");
        }

        // 2、检查是否启用
        if (!mappingTask.isEnable()) {
            throw new IllegalArgumentException("驱动状态必须为启用！");
        }

        String res = "停止驱动成功！";
        // 分别处理数据迁移和数据同步业务
        String model = mappingTask.getModel();
        switch (model) {
        case MappingConstant.SYNC_ALL:
            // 3、移除正在处理的同步任务
            boolean removeRunning = manager.removeRunning(id);
            boolean removeQueue = manager.removeQueue(id);
            res = removeRunning || removeQueue ? res : "没有驱动可以停止！";
            break;
        case MappingConstant.SYNC_INCREMENT:
            Map<String, Map<String, String>> policy = mappingTask.getPolicy();
            Map<String, String> inc = policy.get(MappingConstant.POLICY_INCREMENT);
            if (inc == null) {
                policy.put(MappingConstant.POLICY_INCREMENT, new HashMap<String, String>());
                inc = policy.get(MappingConstant.POLICY_INCREMENT);
            }
            inc.put("enable", "false");
            boolean store = data.saveMapping(id, mappingTask);
            if (!store) {
                logger.error("Stop driver failed!");
                throw new Exception("停止驱动失败!");
            }
            break;
        default:
            break;
        }
        // 触发停止事件
        event.notice(EventConstant.STOP, id);
        return res;
    }

    @Override
    public String delete(String id) throws Exception {
        MappingTask mappingTask = data.getMapping(id);
        if (mappingTask.isEnable()) {
            throw new Exception("请先禁用驱动!");
        }
        // 移除正在运行的任务
        manager.removeRunning(id);
        manager.removeQueue(id);
        boolean remove = data.removeMapping(id);
        if (!remove) {
            logger.error("Delete driver failed!");
            throw new Exception("删除驱动失败!");
        }
        // 触发删除事件
        event.notice(EventConstant.DELTED, id);
        return "删除驱动成功!";
    }

    @Override
    public MappingTask getDriver(String id) throws Exception {
        return data.getMapping(id);
    }

    @Override
    public List<MappingTask> getDriverAll() throws Exception {
        List<MappingTask> list = new ArrayList<MappingTask>();
        Map<String, Object> mps = data.getMappings();
        for (Map.Entry<String, Object> entry : mps.entrySet()) {
            list.add((MappingTask) entry.getValue());
        }
        for (MappingTask task : list) {
            Date d = task.getStartDate();
            // 获取时间差
            task.setStartDateStr(null == d ? "0天0小时0分0秒" : this.getDiff(d.getTime()));

            // 检测驱动可用性
            this.isAlive(task);

            // 获取驱动运行状态
            this.getDriverRunState(task);
        }

        // 降序排序
        Collections.sort(list, new Comparator<MappingTask>() {
            @Override
            public int compare(MappingTask o1, MappingTask o2) {
                if (null == o1.getUpdateTime() || null == o2.getUpdateTime()) {
                    return 0;
                }
                return o2.getUpdateTime().compareTo(o1.getUpdateTime());
            }
        });
        return list;
    }

    @Override
    public Page getDriver(List<MappingTask> list, String driverName, int pageIndex, int pageSize) throws Exception {
        Page page = null;
        if (null != list && !list.isEmpty()) {
            // 根据驱动名称模糊搜索
            if (StringUtils.isNotBlank(driverName)) {
                list = list.stream().filter(t -> StringUtils.contains(t.getName(), driverName)).collect(Collectors.toList());
            }
            final int total = list.size();
            page = new Page(pageIndex, pageSize, total);
            // 有效页数范围
            if (page.getPageNumber() >= pageIndex) {
                int start = pageIndex < 2 ? 0 : (pageIndex * pageSize) - pageSize;
                int end = start + pageSize;
                end = end > total ? total : end;
                final int size = end - start;
                List<MappingTask> res = new ArrayList<>(size);
                for (int i = 0; i < size; i++) {
                    res.add(list.get(start++));
                }
                page.setValue(res);
            }
        }
        return page;
    }

    @Override
    public List<String> getDriverLog(String id) {
        FileReader reader = null;
        BufferedReader br = null;
        List<String> list = null;
        try {
            File file = log.getLog(id);
            reader = new FileReader(file);
            br = new BufferedReader(reader);
            list = new LinkedList<>();
            String str = null;
            while ((str = br.readLine()) != null) {
                list.add(str);
            }
        } catch (Exception e) {
            logger.error(e.getLocalizedMessage());
        } finally {
            log.close(br, reader);
        }
        return list;
    }

    public JSONArray getDriverLogJSONArray(String id) throws JSONException {
        FileReader reader = null;
        BufferedReader br = null;
        JSONArray list = null;
        try {
            File file = log.getLog(id);
            reader = new FileReader(file);
            br = new BufferedReader(reader);
            list = new JSONArray();
            String str = null;
            int i = 0;
            while ((str = br.readLine()) != null) {
                JSONObject obj = new JSONObject(str);
                String _id = obj.getString("id");
                JSONObject _params = obj.getJSONObject("params");
                JSONObject r = new JSONObject();
                r.put("i", i++);
                r.put("id", _id);
                r.put("params", _params);
                list.put(r);
            }
        } catch (Exception e) {
            logger.error(e.getLocalizedMessage());
        } finally {
            log.close(br, reader);
        }
        return list;
    }
    
    @Override
    public File getDriverLogFile(String id) {
        return log.getLog(id);
    }

    @Override
    public void sync(String json) throws Exception {
        // 消息正确性校验,将同步消息转换为JSON格式Object
        JSONObject handle = new JSONObject(json);
        // 获取驱动ID
        String taskId = handle.getString(RestConstant.TASKID);

        // 检查驱动是否存在
        MappingTask driver = getDriver(taskId);
        if (null == driver) {
            throw new IllegalArgumentException("驱动:" + taskId + "不存在.");
        }
        // 检查驱动是否启动
        this.getDriverRunState(driver);
        if (!driver.isRun()) {
            throw new IllegalArgumentException("驱动:" + taskId + "未启动.");
        }

        // 获取同步数据集合
        JSONArray msgArr = handle.getJSONArray(RestConstant.MSG);
        int msgArrLen = msgArr.length();
        if (msgArrLen < 1) {
            throw new IllegalArgumentException("同步的数据为空.");
        }

        // 重新组装同步的数据格式
        JSONArray msg = new JSONArray();
        for (int i = 0; i < msgArrLen; i++) {
            JSONObject d = msgArr.getJSONObject(i);

            // 事件类型
            String eventType = d.getString(RestConstant.EVENTTYPE);
            // 排除非INSERT/UPDATE/DELETE操作的事件
            if (StringUtils.equals(ConnectorConstant.OPERTION_INSERT, eventType) && StringUtils.equals(ConnectorConstant.OPERTION_UPDATE, eventType)
                    && StringUtils.equals(ConnectorConstant.OPERTION_DELETE, eventType)) {
                continue;
            }

            JSONArray data = d.getJSONArray(RestConstant.DATA);
            JSONObject rObj = new JSONObject();
            rObj.put("eventType", eventType);

            //如果是删除操作
            if (StringUtils.equals(ConnectorConstant.OPERTION_DELETE, eventType)) {
                rObj.put("before", data);
                rObj.put("after", new JSONArray());
            } else {
                //如果是新增、修改操作
                rObj.put("before", new JSONArray());
                rObj.put("after", data);
            }
            msg.put(rObj);
        }

        // 数据格式转换
        JSONObject sync = new JSONObject();
        sync.put("taskId", taskId);
        sync.put("msg", msg);

        // 通知manager处理同步消息 
        manager.handle(sync);
    }

    @Override
    public void syncDriverLogAll(String driverId) throws Exception {
        if (StringUtils.isBlank(driverId)) {
            throw new NullPointerException("参数不正确!");
        }
        JSONArray rows = getDriverLogJSONArray(driverId);
        sync(driverId, rows);
    }

    @Override
    public void syncDriverLog(JSONObject msg) throws Exception {
        if (msg.isNull("id") || msg.isNull("rows")) {
            throw new NullPointerException("参数不正确!");
        }
        // 驱动ID
        String driverId = msg.getString("id");
        JSONArray rows = msg.getJSONArray("rows");
        sync(driverId, rows);
    }

    @Override
    public void delDriverLog(JSONObject msg) throws Exception {
        if (msg.isNull("id") || msg.isNull("del")) {
            throw new NullPointerException("参数不正确!");
        }
        log.handle(msg);
    }

    @Override
    public void addDriverLog(JSONObject msg) throws Exception {
        if (msg.isNull("id") || msg.isNull("reason")) {
            throw new NullPointerException("参数不正确!");
        }
        log.handle(msg);
    }

    /**
     * 计算时间差
     * @param begin 开始日期
     * @param end 结束日期
     * @return 例如：10天1小时2分27秒
     */
    private String getDiff(Long beginTime) {
        // 得到两者的毫秒数
        long between = System.currentTimeMillis() - beginTime;
        long day = between / (24 * 60 * 60 * 1000);
        long hour = (between / (60 * 60 * 1000) - day * 24);
        long min = ((between / (60 * 1000)) - day * 24 * 60 - hour * 60);
        long s = (between / 1000 - day * 24 * 60 * 60 - hour * 60 * 60 - min * 60);
        return new StringBuilder().append(day).append("天").append(hour).append("小时").append(min).append("分").append(s).append("秒").toString();
    }

    /**
     * 
     * 查询是否在运行
     * @param id 驱动ID
     * @return true:正在运行，否则false
     */
    private boolean isRun(String id) {
        //判断是否在排队
        Queue<String> queue = manager.getQueue();
        if (queue.contains(id)) {
            return true;
        }
        //判断是否在运行
        Map<String, String> running = manager.getRunning();
        return running.get(id) != null;
    }

    /**
     * 
     * 查询驱动是否可用
     * @param name 驱动名称
     * @return true:正在运行，否则false
     */
    private void isAlive(MappingTask task) {
        try {
            String id = task.getId();
            Map<String, Object> driver = monitor.getDriver(id);
            if (driver == null || driver.isEmpty()) {
                return;
            }
            // 数据源是否可用
            boolean sourceAlive = (boolean) driver.get("source");
            task.getSourceMapping().setAlive(sourceAlive);

            // 目标源是否可用
            boolean targetAlive = (boolean) driver.get("target");
            task.getTargetMapping().setAlive(targetAlive);
        } catch (Exception e) {
            logger.error(e.getLocalizedMessage());
        }
    }

    /**
     * 检查同步属性、主键
     * @param mappingTask
     * @throws IllegalArgumentException
     */
    private void validateSyncFields(MappingTask mappingTask) throws IllegalArgumentException {
        Mapping sourceMapping = mappingTask.getSourceMapping();
        List<Map<String, Object>> sourceFileds = sourceMapping.getFileds();
        if (sourceFileds == null || sourceFileds.isEmpty()) {
            throw new IllegalArgumentException("缺少基本信息!");
        }

        // 目标源驱动
        Mapping targetMapping = mappingTask.getTargetMapping();
        List<Map<String, Object>> fileds = targetMapping.getFileds();
        if (fileds == null || fileds.isEmpty()) {
            throw new IllegalArgumentException("缺少基本信息!");
        }
        // END
    }

    /**
     * 获取驱动运行状态
     * @Title: getDriverRunState 
     * @Description: 获取驱动运行状态
     * @param task
     * @return: void
     */
    private void getDriverRunState(MappingTask task) {
        // 分别处理数据迁移和数据同步业务
        boolean run = false;
        switch (task.getModel()) {
        case MappingConstant.SYNC_ALL:
            // 获取运行状态
            run = this.isRun(task.getId());
            break;
        case MappingConstant.SYNC_INCREMENT:
            // 获取策略
            Map<String, Map<String, String>> policy = task.getPolicy();
            Map<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
            // 是否开启功能
            if (po != null) {
                String enable = po.get("enable");
                run = enable != null && Boolean.valueOf(enable);
            }
            break;
        default:
            break;
        }
        task.setRun(run);
    }

    /**
     * 获取驱动同步事务锁
     * @param id 驱动ID
     * @throws Exception
     */
    private void occupySync(String id) throws Exception {
        if (null != transaction.get(id)) {
            throw new Exception("驱动正在手动同步，请稍后再试。");
        }
        synchronized (transaction) {
            if (null != transaction.get(id)) {
                throw new Exception("驱动正在手动同步，请稍后再试。");
            }
            transaction.put(id, true);
        }
    }

    /**
     * 释放驱动同步事务锁
     * @param id 驱动ID
     * @throws Exception
     */
    private void releaseSync(String id) throws Exception {
        synchronized (transaction) {
            transaction.remove(id);
        }
    }

    private void sync(String driverId, JSONArray rows) throws Exception {
        if (StringUtils.isBlank(driverId) || null == rows) {
            throw new NullPointerException("参数不正确!");
        }
        // 1.获取事务锁
        occupySync(driverId);

        // 2.尝试同步
        JSONArray errRows = new JSONArray();
        JSONArray delRows = new JSONArray();
        try {
            JSONObject row = null;
            int len = rows.length();
            for (int i = 0; i < len; i++) {
                row = rows.getJSONObject(i);

                // 封装同步消息格式
                JSONObject r = new JSONObject();
                r.put("taskId", driverId);
                JSONArray arr = new JSONArray();
                arr.put(row.getJSONObject("params"));
                r.put("msg", arr);

                JSONArray reason = manager.send(r);
                // 同步失败
                if (null != reason) {
                    // 因为是逐条发送
                    errRows.put(reason.get(0));
                    continue;
                }
                // 同步成功
                JSONObject d = new JSONObject();
                d.put("i", row.getString("i"));
                d.put("id", row.getString("id"));
                delRows.put(d);
            }
        } catch (Exception e) {
            logger.error("手动同步日志时异常：", e);
        } finally {
            // 3.删除成功的日志
            if (0 < delRows.length()) {
                JSONObject suc = new JSONObject();
                suc.put("id", driverId);
                suc.put("del", delRows);
                log.handle(suc);
            }
            // 4.关闭事务
            releaseSync(driverId);
        }

        // 5.如果有错误，则抛出异常
        if (0 < errRows.length()) {
            throw new Exception(errRows.toString());
        }
    }
    
}
